#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#define _GNU_SOURCE
#include <stdio.h>

#include "queue.h"

#include "compat.h"

#include "psd_session.h"
#include "psd_conn.h"
#include "psd_protocols.h"
#include "psd_optimize.h"
#include "psd_logger.h"
#include "psd_event.h"
#include "psd_route.h"
#include "psd_auth.h"
#include "psd_path.h"
#include "psd_path_handler.h"
#include "psd_measurement.h"
#include "psd_settings.h"
#include "psd_xfer.h"
#include "psd_user_settings.h"
#include "psd_route_settings.h"
#include "psd_default_settings.h"
#include "psd_path.h"
#include "psd_config.h"
#include "psd_modules.h"
#include "psd_backend_handler.h"

int psd_connect_hop(psdSess *sess, xspHop *curr_child, psdConn **ret_conn, char **ret_error_msg);
int psd_connect_hop_protocol(psdSess *sess, xspHop *curr_child, psdSettings *policy, const char *protocol, psdConn **ret_conn, char **ret_error_msg);

static LIST_HEAD(listhead, psd_session_t) sessions_list;
static pthread_mutex_t sessions_list_lock;
static int session_count;

#ifdef NETLOGGER
int get_next_stream_id() {
  int i;
  for (i=1; i<MAX_ID; i++)
    if (stream_ids[i] == 0) {
      stream_ids[i] = 1;
      return i;
    }
  psd_info(0, "Ran out of stream_ids for NetLogger!\n");
  return 0;
}
#endif

int psd_sessions_init() {
	if (pthread_mutex_init(&sessions_list_lock, 0) != 0)
		goto error_exit;

	session_count = 0;

	LIST_INIT(&sessions_list);

	return 0;

error_exit:
	return -1;
}

int psd_register_session(psdSess *sess) {
	int retval = -1;

	pthread_mutex_lock(&sessions_list_lock);
	{
		psdSess *curr_sess;

		for(curr_sess = sessions_list.lh_first; curr_sess != NULL; curr_sess = curr_sess->sess_list.le_next) {
			if (!memcmp(curr_sess->id, sess->id, 2*XSP_SESSIONID_LEN + 1)) {
				psd_err(5, "tried to re-register session %s", sess->id);
				break;
			}
		}

		if (curr_sess == NULL) {
			psd_info(5, "registering session %s", sess->id);

			LIST_INSERT_HEAD(&sessions_list, sess, sess_list);
			retval = 0;
			session_count++;
		}
	}
	pthread_mutex_unlock(&sessions_list_lock);

	return retval;
}

void psd_unregister_session(psdSess *sess) {

	pthread_mutex_lock(&sessions_list_lock);
	{
		psd_info(5, "unregistering session %s", sess->id);

		if (sess->sess_list.le_prev != NULL) {
			LIST_REMOVE(sess, sess_list);
			session_count--;
		}
	}
	pthread_mutex_unlock(&sessions_list_lock);
}

psdSess *psd_lookup_session(char *sess_id) {
	psdSess *curr_sess;

	psd_info(5, "finding session %s", sess_id);

	pthread_mutex_lock(&sessions_list_lock);
	{
		for(curr_sess = sessions_list.lh_first; curr_sess != NULL; curr_sess = curr_sess->sess_list.le_next) {
			if (!memcmp(curr_sess->id, sess_id, 2*XSP_SESSIONID_LEN + 1)) {
				psd_info(5, "found session %s", sess_id);
				break;
			}
		}

		if (curr_sess) {
			__psd_session_get_ref(curr_sess);
		}
	}
	pthread_mutex_unlock(&sessions_list_lock);

	return curr_sess;
}

psdSess **psd_get_sessions(int *count) {
	psdSess *curr_sess;
	psdSess **sess_list;

	pthread_mutex_lock(&sessions_list_lock);
	{
		sess_list = malloc(sizeof(psdSess *) * session_count);
		if (sess_list) {
			int i = 0;
			for(curr_sess = sessions_list.lh_first; curr_sess != NULL; curr_sess = curr_sess->sess_list.le_next) {
				sess_list[i] = __psd_session_get_ref(curr_sess);
				i++;
			}

			*count = session_count;
		}
	}
	pthread_mutex_unlock(&sessions_list_lock);

	return sess_list;
}

psdSess *psd_session_get_ref(psdSess *sess) {
	psdSess *ret_sess;

	pthread_mutex_lock(&sessions_list_lock);
	{
		ret_sess = __psd_session_get_ref(sess);
	}
	pthread_mutex_unlock(&sessions_list_lock);

	return ret_sess;
}

psdSess *__psd_session_get_ref(psdSess *sess) {
	pthread_mutex_lock(&sess->references_lock);
	{
		psd_info(5, "%s: got reference for session", sess->id);
		sess->references++;
	}
	pthread_mutex_unlock(&sess->references_lock);

	return sess;
}

void psd_session_put_ref(psdSess *sess) {

	pthread_mutex_lock(&sess->references_lock);

	psd_info(5, "%s: put reference for session", sess->id);
	sess->references--;

	if (sess->references == 0) {
		psd_info(5, "%s: no more references for session, cleaning up", sess->id);
		psd_end_session(sess);
	} else {
		pthread_mutex_unlock(&sess->references_lock);
	}
}

int psd_setup_session(psdSess *sess, char ***ret_error_msgs) {
	int i;

	for(i = 0; i < sess->child_count; i++) {
		const char *new_hop_id;
		xspHop *curr_child = sess->child[i];

		new_hop_id = psd_route_lookup(xsp_hop_getid(curr_child));
		if (new_hop_id && strcmp(new_hop_id, "direct") != 0 && strcmp(new_hop_id, xsp_hop_getid(curr_child)) != 0) {
			xspHop *new_hop;

			psd_info(0, "using \"%s\" as a hop to \"%s\"", new_hop_id, xsp_hop_getid(curr_child));

			new_hop = xsp_alloc_hop();
			if (new_hop) {
				new_hop->child = malloc(sizeof(struct xsp_hop *));
				if (!new_hop->child) {
					free(new_hop);
				} else {
					strncpy(new_hop->hop_id, new_hop_id, sizeof(new_hop->hop_id));
					new_hop->child[0] = sess->child[i];
					new_hop->child_count = 1;
//					new_hop->flags = XSP_HOP_NATIVE | XSP_UNNECESSARY;
					new_hop->flags = XSP_HOP_NATIVE;
					new_hop->session = sess;
					sess->child[i] = new_hop;
				}
			}
		}
	}

	char **msgs = malloc(sizeof(char *) * sess->child_count);
	if (!msgs) {
		psd_err(0, "Couldn't allocate space to hold child error messages");
		return -1;
	}

	*ret_error_msgs = msgs;

	for(i = 0; i < sess->child_count; i++) {
		xspHop *curr_child = sess->child[i];
		psdConn *new_conn;
		char *error_msg;

		if (psd_connect_hop(sess, curr_child, &new_conn, &error_msg) != 0) {
			psd_event("phoebus.connection.failed", curr_child, "HOP=%s", xsp_hop_getid(curr_child));
			psd_err(0, "couldn't connect to \"%s\"", xsp_hop_getid(curr_child));
			msgs[i] = error_msg;

			if (curr_child->flags & XSP_UNNECESSARY) {
				if (curr_child->child != NULL) {
					xspHop **new_child_list;

					psd_info(0, "reparenting children of %s", xsp_hop_getid(curr_child));

					new_child_list = realloc(sess->child, sizeof(xspHop *) * (sess->child_count + curr_child->child_count));
					if (new_child_list) {
						char **new_msgs;
						int j;

						for(j = 0; j < curr_child->child_count; j++) {
							new_child_list[sess->child_count + j] = curr_child->child[j];
						}

						psd_info(0, "%d children of %s reparented", curr_child->child_count, xsp_hop_getid(curr_child));

						sess->child = new_child_list;
						sess->child_count = sess->child_count + curr_child->child_count;

						free(curr_child->child);
						curr_child->child_count = 0;
						curr_child->child = NULL;

						new_msgs = realloc(msgs, sizeof(char *) * sess->child_count);
						if (new_msgs) {
							msgs = new_msgs;
							*ret_error_msgs = msgs;
						}
					}
				}
			}
		} else {
			psd_info(5, "completed connection to \"%s\"" , xsp_hop_getid(curr_child));
			LIST_INSERT_HEAD(&(sess->child_conns), new_conn, sess_entries);
			gettimeofday(&(new_conn->start_time), NULL);
			psd_event("phoebus.connection.start", new_conn, "HOP=%s PROTOCOL=%s", xsp_hop_getid(curr_child), new_conn->protocol);
			msgs[i] = NULL;
		}
	}

	if (psd_register_session(sess) != 0) {
		*ret_error_msgs = NULL;
		goto error_exit;
	}

	return 0;

error_exit:
	return -1;
}

void psd_end_session(psdSess *sess) {
	psd_unregister_session(sess);

	psd_session_close_connections(sess);

	psd_free_session(sess);
}

psdSess *psd_convert_xspSess(xspSess *old_sess) {
	psdSess *new_sess;
	int i;

	new_sess = psd_alloc_sess();
	if (!new_sess) {
		return NULL;
	}

	// copy all the old data over
	bcopy(old_sess->sess_id, new_sess->id, 2*XSP_SESSIONID_LEN+1);
	new_sess->child = old_sess->child;
	new_sess->child_count = old_sess->child_count;

	// initialize the new data
	new_sess->references = 1;

	for(i = 0; i < new_sess->child_count; i++) {
		new_sess->child[i]->session = (xspSess *) new_sess;
	}

	LIST_INIT(&new_sess->parent_conns);
	LIST_INIT(&new_sess->child_conns);
	LIST_INIT(&new_sess->parent_data_conns);
        LIST_INIT(&new_sess->child_data_conns);

	return new_sess;
}

psdSess *psd_alloc_sess() {
	psdSess *new_sess;

	new_sess = (psdSess *) malloc(sizeof(psdSess));
	if (!new_sess)
		goto error_exit;

	bzero(new_sess, sizeof(*new_sess));

	if (pthread_mutex_init(&(new_sess->references_lock), 0) < 0)
		goto error_exit2;

#ifdef NETLOGGER
	new_sess->nl_id = get_next_stream_id();
#endif
	return new_sess;

error_exit2:
	free(new_sess);
error_exit:
	return NULL;
}

inline char *psd_session_get_user(psdSess *sess) {
        return sess->user;
}

inline void psd_session_set_user(psdSess *sess, char *user) {
	sess->user = user;
}

inline char *psd_session_get_id(psdSess *sess) {
	return sess->id;
}

inline void psd_session_close_connections(psdSess *sess) {
	psdConn *curr_conn;

	psd_info(5, "%s: closing connections", sess->id);

	// shutdown the parent connections
	LIST_FOREACH(curr_conn, &(sess->parent_conns), sess_entries) {
		psd_conn_shutdown(curr_conn, (PSD_SEND_SIDE | PSD_RECV_SIDE));
		if (curr_conn->path != NULL) {
			curr_conn->path->close_channel(curr_conn->path, curr_conn->channel);
		}
	}

	// shutdown the children connections
	LIST_FOREACH(curr_conn, &(sess->child_conns), sess_entries) {
		psd_conn_shutdown(curr_conn, (PSD_SEND_SIDE | PSD_RECV_SIDE));
		if (curr_conn->path != NULL) {
			curr_conn->path->close_channel(curr_conn->path, curr_conn->channel);
		}
	}
}

void psd_free_session(psdSess *sess) {
	psdConn *curr_conn, *next_conn;

	psd_info(5, "%s: freeing session", sess->id);

	while(!LIST_EMPTY(&sess->parent_conns)) {
		curr_conn = LIST_FIRST(&sess->parent_conns);
		LIST_REMOVE(curr_conn, sess_entries);
		psd_conn_free(curr_conn);
	}

	while(!LIST_EMPTY(&sess->child_conns)) {
		curr_conn = LIST_FIRST(&sess->child_conns);
		LIST_REMOVE(curr_conn, sess_entries);
		psd_conn_free(curr_conn);
	}

	if (sess->child) {
		int i;

		for(i = 0; i < sess->child_count; i++) {
			xsp_free_hop(sess->child[i], 1);
		}

		free(sess->child);
	}

	sess->credentials->free(sess->credentials);

	if (sess->user)
		free(sess->user);

	pthread_mutex_destroy(&(sess->references_lock));

#ifdef NETLOGGER
	stream_ids[sess->nl_id] = 0;
#endif
	free(sess);
}

int psd_num_sessions() {
	int count;
	psdSess *curr_sess;

	count = 0;

	pthread_mutex_lock(&sessions_list_lock);
	{
		for(curr_sess = sessions_list.lh_first; curr_sess != NULL; curr_sess = curr_sess->sess_list.le_next)
			count++;
	}
	pthread_mutex_unlock(&sessions_list_lock);

	return count;
}

int psd_session_get_stat(psdSess *sess, uint16_t type, void *optval, size_t *optlen) {
	int retval = -1;

	switch(type) {
		case PSD_STATS_BYTES_READ:
			if (*optlen >= sizeof(uint64_t)) {
				psdConn *conn;
				uint64_t total_bytes;

				total_bytes = 0;

				LIST_FOREACH(conn, &(sess->parent_conns), sess_entries) {
					uint64_t bytes_read = 0;
					size_t bytes_read_size = sizeof(bytes_read);

					if (psd_conn_get_stat(conn, PSD_STATS_BYTES_READ, &bytes_read, &bytes_read_size) == 0) {
						total_bytes += bytes_read;
					}
				}

				LIST_FOREACH(conn, &(sess->child_conns), sess_entries) {
					uint64_t bytes_read = 0;
					size_t bytes_read_size = sizeof(bytes_read);

					if (psd_conn_get_stat(conn, PSD_STATS_BYTES_READ, &bytes_read, &bytes_read_size) == 0) {
						total_bytes += bytes_read;
					}
				}

				*((uint64_t *)optval) = total_bytes;
				*optlen = sizeof(total_bytes);
				retval = 0;
			}
			break;

		case PSD_STATS_BYTES_WRITTEN:
			if (*optlen >= sizeof(uint64_t)) {
				psdConn *conn;
				uint64_t total_bytes;

				total_bytes = 0;

				LIST_FOREACH(conn, &(sess->parent_conns), sess_entries) {
					uint64_t bytes_written = 0;
					size_t bytes_written_size = sizeof(bytes_written);

					if (psd_conn_get_stat(conn, PSD_STATS_BYTES_WRITTEN, &bytes_written, &bytes_written_size) == 0) {
						total_bytes += bytes_written;
					}
				}

				LIST_FOREACH(conn, &(sess->child_conns), sess_entries) {
					uint64_t bytes_written = 0;
					size_t bytes_written_size = sizeof(bytes_written);

					if (psd_conn_get_stat(conn, PSD_STATS_BYTES_WRITTEN, &bytes_written, &bytes_written_size) == 0) {
						total_bytes += bytes_written;
					}
				}


				*((uint64_t *)optval) = total_bytes;
				*optlen = sizeof(total_bytes);
				retval = 0;
			}
			break;
	}

	return retval;
}

/*
 *  int psd_session_main_loop(psdSess *sess):
 *      This function spawns the I/O threads for the given session and waits
 *      until they exit before returning.
 */
int psd_session_main_loop(psdSess *sess) {
	psdBackendHandler *xfer_backend;
	psdXferArgs *p_args, *c_args;
	char *transfer_backend = "sr";
	char *str_val;
	int ret;

	if (psd_depot_settings_get("generic", "transfer_backend", &str_val) == 0) {
		transfer_backend = str_val;
	}

	if ((xfer_backend = psd_get_backend(transfer_backend)) == NULL) {
		psd_err(0, "Transfer backend module not loaded: %s", transfer_backend);
		goto error_exit;
	}
	
	if (LIST_NEXT(LIST_FIRST(&sess->parent_conns), sess_entries) != NULL || LIST_NEXT(LIST_FIRST(&sess->child_conns), sess_entries) != NULL) {
		psd_err(0, "can't handle sessions with multiple parents or children, yet!");
		goto error_exit;
	}

	p_args = psd_alloc_xfer_args_set(sess, LIST_FIRST(&sess->parent_conns), LIST_FIRST(&sess->child_conns), PSD_BUF_SIZE);
	if (!p_args) {
		psd_err(5, "couldn't allocate transfer structure");
		goto error_exit;
	}

	c_args = psd_alloc_xfer_args_set(sess, LIST_FIRST(&sess->child_conns), LIST_FIRST(&sess->parent_conns), PSD_BUF_SIZE);
	if (!c_args) {
		psd_err(5, "couldn't allocate transfer structure");
		goto error_exit1;
	}

	psd_info(10, "Starting transfer backend %s for session %s", transfer_backend, psd_session_get_id(sess));
	
	if (xfer_backend->run(p_args, c_args, -1) != 0) {
		psd_err(0, "Could not start transfer backend");
		goto error_exit2;
	}
	
	psd_free_xfer_args(p_args);
	psd_free_xfer_args(c_args);

	return 0;
	
	
	/*
	// decide which backend to use	
	if (strcmp("sr", transfer_backend) == 0) {
		xfer_func_0 = xfer_func_1 = &psd_sr_xfer_thread;
	}
	else if (strcmp("slabs-ingress", transfer_backend) == 0) {
#ifdef HAVE_SLABS
		// buffers should be sized according to the bulk-data transfer direction...
		psd_info(11, "Initializing SLaBS buffer");

		args0->slab = malloc(sizeof(struct ring_buffer));
		//args1->slab = malloc(sizeof(struct ring_buffer));

		ring_buffer_create(args0->slab, PSD_SLAB_ORDER);
		//ring_buffer_create(args1->slab, PSD_SLAB_ORDER);

		psd_info(11, "Created buffer of size %lu", ring_buffer_count_free_bytes(args0->slab));
                xfer_func_0 = &psd_slabs_ingress_xfer_thread;
		xfer_func_1 = &psd_sr_xfer_thread;

		psd_register_xfer_arg(args0);
		
		// we know we're ready to receive, wait for message from other side
		msg = psd_conn_get_msg(args1->src, 0);
		if (!msg) {
			psd_err(0, "did not get XFER ACK");
			goto error_exit2;
		}
		if (msg->type != XSP_MSG_SESS_ACK) {
			psd_err(0, "did not get ACK as expected");
			goto error_exit2;
		}
		else
			psd_info(10, "received XFER ACK");
			
#else
		psd_err(0, "SLaBS backend was not enabled, run configure with --enable-slabs");
		psd_info(0, "Using send/recv backend!");
		xfer_func_0 = xfer_func_1 = &psd_sr_xfer_thread;
#endif
        }
	else if (strcmp("slabs-egress", transfer_backend) == 0) {
#ifdef HAVE_SLABS
                // slabs buffer should be enabled according to the bulk-data transfer direction...
                psd_info(11, "Initializing SLaBS buffer");

                args0->slab = malloc(sizeof(struct ring_buffer));
                //args1->slab = malloc(sizeof(struct ring_buffer));

                ring_buffer_create(args0->slab, PSD_SLAB_ORDER);
                //ring_buffer_create(args1->slab, PSD_SLAB_ORDER);

                psd_info(11, "Created buffer of size %lu", ring_buffer_count_free_bytes(args0->slab));
                xfer_func_0 = &psd_slabs_egress_xfer_thread;
		xfer_func_1 = &psd_sr_xfer_thread;

		psd_register_xfer_arg(args0);

		// buf created and registered, signal the ingress GW
		if (psd_conn_send_msg(args1->sink, XSP_MSG_SESS_ACK, 0) <= 0) {
			psd_err(0, "could not send XFER ACK");
			goto error_exit2;
		}
		else
			psd_info(10, "sent XFER ACK");
#else
                psd_err(0, "SLaBS backend was not enabled, run configure with --enable-slabs");
                psd_info(0, "Using send/recv backend!");
                xfer_func_0 = xfer_func_1 = &psd_sr_xfer_thread;
#endif
	}
	else if ((strcmp("splice", transfer_backend) == 0) &&
		 (strcmp("TCP", args0->src->protocol) == 0) &&
		 (strcmp("TCP", args1->src->protocol) == 0)) {

		xfer_func_0 = xfer_func_1 = &psd_splice_xfer_thread;

		// setup splice pipes
		ret = pipe(args0->pipes);
		
		if (ret < 0) {
			psd_err(0, "pipe failed: %s", strerror(errno));
			goto error_exit1;
		}
		

		ret = pipe(args1->pipes);
		
		if (ret < 0) {
			psd_err(0, "pipe failed: %s", strerror(errno));
			goto error_exit1;
		}

		// start the sink threads

		pthread_mutex_lock(&(args0->sink_lock));
		// create an I/O thread
		if (pthread_create(&sink_thr1, NULL, psd_splice_xfer_sink, (void *) args0) < 0) {
			pthread_mutex_unlock(&(args0->sink_lock));
			goto error_exit2;
		}

		// wait for the child to initialize
		pthread_cond_wait(&(args0->sink_cond), &(args0->sink_lock));
		pthread_mutex_unlock(&(args0->sink_lock));


		pthread_mutex_lock(&(args1->sink_lock));
		// create an I/O thread
		if (pthread_create(&sink_thr2, NULL, psd_splice_xfer_sink, (void *) args1) < 0) {
			pthread_mutex_unlock(&(args1->sink_lock));
			goto error_exit2;
		}
		
		// wait for the child to initialize
		pthread_cond_wait(&(args1->sink_cond), &(args1->sink_lock));
		pthread_mutex_unlock(&(args1->sink_lock));

		splice = 1;
	}
	else {
		xfer_func_0 = xfer_func_1 = &psd_sr_xfer_thread;
	}
	
	pthread_mutex_lock(&(args0->lock));
	// create an I/O thread
   	if (pthread_create(&thread1, NULL, xfer_func_0, (void *) args0) < 0) {
   		pthread_mutex_unlock(&(args0->lock));
   		goto error_exit2;
	}

	// wait for the child to initialize
	pthread_cond_wait(&(args0->cond), &(args0->lock));
	pthread_mutex_unlock(&(args0->lock));
	
	// create an I/O thread
	pthread_mutex_lock(&(args1->lock));
	
   	if (pthread_create(&thread2, NULL, xfer_func_1, (void *) args1) < 0) {
   		pthread_mutex_unlock(&(args1->lock));
   		goto error_exit3;
	}
	// wait for the child to initialize
	pthread_cond_wait(&(args1->cond), &(args1->lock));
	pthread_mutex_unlock(&(args1->lock));

	// setup the threads to transfer "infinitely"
	pthread_mutex_lock(&(args0->lock));
	pthread_mutex_lock(&(args1->lock));
	{
		args0->xfer_amt = -1;
		pthread_cond_signal(&(args0->cond));

		args1->xfer_amt = -1;
		pthread_cond_signal(&(args1->cond));
	}
	pthread_mutex_unlock(&(args0->lock));
	pthread_mutex_unlock(&(args1->lock));

	if (splice) {
		pthread_mutex_lock(&(args0->sink_lock));
		pthread_mutex_lock(&(args1->sink_lock));
		{
			args0->xfer_amt = -1;
			pthread_cond_signal(&(args0->sink_cond));
			
			args1->xfer_amt = -1;
			pthread_cond_signal(&(args1->sink_cond));
		}
		pthread_mutex_unlock(&(args0->sink_lock));
		pthread_mutex_unlock(&(args1->sink_lock));
	}

	// wait for the I/O threads to finish
	pthread_join(thread1, NULL);
	pthread_join(thread2, NULL);

	if (splice) {
		pthread_join(sink_thr1, NULL);
		pthread_join(sink_thr2, NULL);
		// pipes are closed in the threads
	}

#ifdef NETLOGGER
	psd_info(0, "finalizing NL summarizer.");
        NL_transfer_finalize(prog_summ);
#endif

	*/

 error_exit2:
	psd_free_xfer_args(c_args);
 error_exit1:
	psd_free_xfer_args(p_args);
 error_exit:
	return -1;
}

void psd_session_finalize(psdSess *sess) {
	uint64_t bytes_written;
	size_t bytes_written_size;

	bytes_written_size = sizeof(bytes_written);

	if (psd_session_get_stat(sess, PSD_STATS_BYTES_WRITTEN, &bytes_written, &bytes_written_size) == 0) {
		psd_log_measurement_uint(sess->end_time, psd_session_get_id(sess), "TransferSize", bytes_written);
		psd_log_measurement_dbl(sess->end_time, psd_session_get_id(sess), "TransferRate", (bytes_written/difftv(&sess->start_time, &sess->end_time))*8/1000/1000);
		psd_log_measurement_dbl(sess->end_time, psd_session_get_id(sess), "TransferTime", difftv(&sess->start_time, &sess->end_time));
	}
}

psdConn *psd_connect_hop_control(const char *hop_id) {
	char *hostname, *port_str;
	int port;
	psdConn *conn;
	psdSettings *tcp_settings;

	if (xsp_parse_hopid(hop_id, &hostname, &port_str) != 0) {
		psd_err(0, "invalid hop id: %s", hop_id);
		goto error_exit;
	}

	tcp_settings = psd_settings_alloc();
	if (!tcp_settings) {
		psd_err(0, "couldn't allocate tcp protocol settings");
		goto error_exit_parsed;
	}

	sscanf(port_str, "%d", &port);

	if (psd_settings_set_int_2(tcp_settings, "tcp", "port", port) != 0) {
		psd_err(0, "couldn't set TCP port");
		goto error_exit_settings;
	}

	conn = psd_protocol_connect_host(hostname, "tcp", tcp_settings);
	if (!conn) {
		psd_err(0, "couldn't connect to %s on port %d with tcp", hostname, port);
		goto error_exit_settings;
	}

	return conn;

error_exit_settings:
	psd_settings_free(tcp_settings);
error_exit_parsed:
	free(hostname);
	free(port_str);
error_exit:
	return NULL;
}

#if 0
int psd_connect_endhost(psdSess *sess, xspHop *curr_child, psdConn **ret_conn) {

}

void *psd_handle_control_conn(void *arg) {
	xspMsg *msg;
	psdConn *conn = (psdConn *) arg;
	int authenticated;
	int have_session;

	authenticated = 0;

	do {
		// before authentication, we'll respond to pings and authentication requests only

		msg = psd_conn_get_msg(new_conn, 0);
		if (!msg) {
			psd_err(5, "Did not receive properly formed message");
			goto error_exit;
		}

		switch(msg->type) {
			case XSP_MSG_PING:
				{
					psd_conn_send_msg(new_conn, XSP_MSG_PONG, NULL);
				}
				break;

			case XSP_MSG_AUTH_TYPE:
				{
					xspAuthType *auth_type = msg->msg_body;
					if (psd_authenticate_connection(new_conn, auth_type->name, &credentials) != 0) {
						psd_err(0, "Authentication failed");
						xsp_free_msg(msg);
						goto error_exit;
					}

					authenticated = 1;
				}
				break;

			default:
				{
					psd_err(0, "Invalid message received");
					goto error_exit;
				}
				break;
		}

		xsp_free_msg(msg);
	} while(authenticated == 0)

	have_session = 0;
	do {

		msg = psd_conn_get_msg(new_conn, 0);
		if (!msg) {
			psd_err(5, "Did not receive properly formed message");
			goto error_exit;
		}

		switch(msg->type) {
			case XSP_MSG_PING:
				{
					psd_conn_send_msg(new_conn, XSP_MSG_PONG, NULL);
				}
				break;

			case XSP_MSG_SESS_OPEN:
				{
					have_session = 1;
				}
				break;

			default:
				{
					psd_err(0, "Invalid message received");
					xsp_free_msg(msg);
					goto error_exit_credentials;
				}
				break;
		}

		xsp_free_msg(msg);
	} while (have_session == 0);

	have_data_parameters = 0;
	do {
		msg = psd_conn_get_msg(new_conn, 0);
		if (!msg) {
			psd_err(5, "Did not receive properly formed message");
			goto error_exit;
		}

		switch(msg->type) {
			case XSP_MSG_PING:
				{
					psd_conn_send_msg(new_conn, XSP_MSG_PONG, NULL);
				}
				break;

			case XSP_MSG_DATA_CHANNEL_REQUEST:
				{
					// XXX we need a way to describe the 'limits'. Should be compared with bwctl.
				}
				break;

			default:
				{
					psd_err(0, "Invalid message received");
					xsp_free_msg(msg);
					goto error_exit_credentials;
				}
				break;
		}

		xsp_free_msg(msg);
	} while (have_data_parameters == 0);

	return NULL;

error_exit_credentials:
	psd_credentials_free(credentials);
error_exit:
	psd_conn_close(conn);
	psd_conn_free(conn);

	return NULL;
}
#endif

int psd_connect_hop(psdSess *sess, xspHop *curr_child, psdConn **ret_conn, char **ret_error_msg) {
	int connected;
	psdConn *new_conn = NULL;
	psdSettings *settings = NULL;
	char **protocols;
	int protocol_count;
	char **path_protocols;
	int path_protocol_count;
	int n;
	char *path_id;
	int i;
	psdPath *path = NULL;
	psdChannel *channel = NULL;
	int bandwidth;
	char *error_msg = NULL;

	if (psd_get_settings(sess, curr_child, &settings) != 0) {
		psd_err(5, "couldn't get route settings");
		goto error_exit;
	}

	if (psd_run_optimizers(sess, xsp_hop_getid(curr_child), settings) != 0) {
		psd_warn(5, "optimization failed for connection: %s", xsp_hop_getid(curr_child));
	}

	connected = 0;

	n = psd_settings_get_list(settings, "protocols", &protocols, &protocol_count);
	if (n != 0) {
		psd_err(0, "settings does not contain a protocol list, using default");

		protocols = psd_get_protocol_list(&protocol_count);
		if (!protocols) {
	 		asprintf(ret_error_msg, "Internal error: configured path %s doesn't exist", path_id);
			psd_err(0, "couldn't load default protocol list");
			goto error_exit_settings;
		}
	}

	if (psd_settings_get_list(settings, "path_protocols", &path_protocols, &path_protocol_count) != 0) {
		psd_info(0, "no path to instantiate to get to %s", xsp_hop_getid(curr_child));
		path_protocols = NULL;
		path_protocol_count = 0;
	}

	if (psd_settings_get_int(settings, "bandwidth", &bandwidth) != 0) {
		psd_info(0, "no path to instantiate to get to %s", xsp_hop_getid(curr_child));
		bandwidth = 0;
	}

	for(i = 0; i < path_protocol_count; i++) {
 		char *error_msg;

		if (psd_get_path(path_protocols[i], settings, &path, &error_msg) != 0) {
			if (i < path_protocol_count - 1) {
				continue;
			}

			if (ret_error_msg)
				*ret_error_msg = error_msg;
			else 
				error_msg = "Unknown error";

			psd_err(0, "%s: couldn't get path information: %s", xsp_hop_getid(curr_child), error_msg);
			goto error_exit_protocols;
		}

		if (path->new_channel(path, bandwidth, &channel, &error_msg) != 0) {
			if (i < path_protocol_count - 1) {
				free(error_msg);
				continue;
			}

			if (ret_error_msg)
				*ret_error_msg = error_msg;
			else 
				error_msg = "Unknown error";

			psd_err(0, "%s: couldn't allocate a channel: %s", xsp_hop_getid(curr_child), error_msg);
			goto error_exit_protocols;
		}
	}

	for(i = 0; i < protocol_count; i++) {
 		// get rid of the previous instance, we should probably keep a list of strings
 		if (error_msg) {
 			free(error_msg);
 			error_msg = NULL;
 		}

		if (psd_connect_hop_protocol(sess, curr_child, settings, protocols[i], &new_conn, &error_msg) == 0) {
			break;
		}
	}

	if (i == protocol_count) {
 		*ret_error_msg = error_msg;
		psd_err(0, "couldn't connect to %s using any protocols", xsp_hop_getid(curr_child));
		goto error_exit_channel;
	}

	new_conn->settings = settings;

	// associate the new connection with the given path/channel
	if (channel) {
		new_conn->path = path;
		new_conn->channel = channel;
		LIST_INSERT_HEAD(&(channel->connlist), new_conn, channel_entries);
	}

	strlist_free(protocols, protocol_count);

	*ret_conn = new_conn;

	return 0;

error_exit_channel:
	if (channel != NULL) {
		path->close_channel(path, channel);
	}
error_exit_protocols:
	strlist_free(protocols, protocol_count);
error_exit_settings:
	psd_settings_free(settings);
error_exit:
	return -1;
}

int psd_connect_hop_protocol(psdSess *sess, xspHop *curr_child, psdSettings *settings, const char *protocol, psdConn **ret_conn, char **ret_error_msg) {
	psdConn *new_conn;
	xspMsg *msg;
	char *hostname, *port_str;
	int port;


	if (xsp_parse_hopid(xsp_hop_getid(curr_child), &hostname, &port_str) != 0) {
		psd_err(0, "invalid hop id");
		goto error_exit;
	}

	sscanf(port_str, "%d", &port);

	if (psd_settings_set_int_2(settings, protocol, "port", port) != 0) {
		psd_err(0, "couldn't set UDT port");
		goto error_exit_parsed;
	}

	new_conn = psd_protocol_connect_host(hostname, protocol, settings);
	if (!new_conn) {
 		asprintf(ret_error_msg, "Couldn't connect to \"%s\" using protocol \"%s\"", xsp_hop_getid(curr_child), protocol);
		psd_err(5, "connect() to \"%s\" using \"%s\" protocol failed", xsp_hop_getid(curr_child), protocol);
		goto error_exit_parsed;
	}

	new_conn->session = sess;

	if (curr_child->flags & XSP_HOP_NATIVE) {
		new_conn->session = sess;

		// if we aren't connecting to the end host, setup a full session or
		// peering session
		
		if (psd_request_authentication(sess, new_conn, "ANON") != 0) {
 			asprintf(ret_error_msg, "Authenictation failed for hop \"%s\"", xsp_hop_getid(curr_child));
			psd_err(5, "authorization failed for: %s", xsp_hop_getid(curr_child));
			goto error_exit_conn;
		}

		if (psd_conn_send_msg(new_conn, XSP_MSG_SESS_OPEN, curr_child) < 0) {
 			asprintf(ret_error_msg, "Connection to \"%s\" closed before we could send a session open message",  xsp_hop_getid(curr_child));
			psd_err(5, "send message to %s failed", xsp_hop_getid(curr_child));
			goto error_exit_conn;
		}

		msg = psd_conn_get_msg(new_conn, 0);
		if (!msg) {
 			asprintf(ret_error_msg, "Connection to \"%s\" did not respond after we sent a session open message",  xsp_hop_getid(curr_child));
			psd_err(5, "recv message from %s failed", xsp_hop_getid(curr_child));
			goto error_exit_conn;
		}

		if (msg->type == XSP_MSG_SESS_NACK) {
 			*ret_error_msg = strdup((char *) msg->msg_body);
			psd_err(5, "received session nack message from %s: %s", xsp_hop_getid(curr_child), (char *) msg->msg_body);
			goto error_exit_msg;
		} else if (msg->type != XSP_MSG_SESS_ACK) {
 			asprintf(ret_error_msg, "Connection to \"%s\" did not send us a session ack or nack in response to the session open message",  xsp_hop_getid(curr_child));
			psd_err(5, "received non-ack message from %s", xsp_hop_getid(curr_child));
			goto error_exit_msg;
		}

		psd_conn_set_session_status(new_conn, 1);

		xsp_free_msg(msg);
	}

	free(hostname);
	free(port_str);

	*ret_conn = new_conn;

	return 0;

error_exit_msg:
	xsp_free_msg(msg);
error_exit_conn:
	psd_conn_shutdown(new_conn, (PSD_RECV_SIDE | PSD_SEND_SIDE));
error_exit_parsed:
	free(hostname);
	free(port_str);
error_exit:
	return -1;
}

int psd_get_settings(psdSess *sess, xspHop *curr_child, psdSettings **ret_settings) {
	psdSettings *default_settings = NULL;
	psdSettings *route_settings = NULL;
	psdSettings *user_settings = NULL;
	psdSettings *new_settings = NULL;
	psdSettings *settings = NULL;
	int oscars_src_tagged = 0;
	
	settings = psd_settings_alloc();

	default_settings = psd_default_settings(PSD_BOTH);
	if (default_settings) {
		new_settings = psd_settings_merge(settings, default_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge default settings with user settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);

		settings = new_settings;
	}

	default_settings = psd_default_settings(PSD_OUTGOING);
	if (default_settings) {
		new_settings = psd_settings_merge(settings, default_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge default settings with user settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);

		settings = new_settings;
	}

	user_settings = psd_user_settings(sess->credentials->get_user(sess->credentials), PSD_BOTH);
	if (user_settings) {
		new_settings = psd_settings_merge(settings, user_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge default settings with user settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);

		settings = new_settings;
	}

	user_settings = psd_user_settings(sess->credentials->get_user(sess->credentials), PSD_OUTGOING);
	if (user_settings) {
		new_settings = psd_settings_merge(settings, user_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge default settings with user settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);

		settings = new_settings;
	}

	route_settings = psd_route_settings_lookup(xsp_hop_getid(curr_child), PSD_BOTH);
	if (route_settings) {
		new_settings = psd_settings_merge(settings, route_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge settings with route settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);
		settings = new_settings;
	}

	route_settings = psd_route_settings_lookup(xsp_hop_getid(curr_child), PSD_OUTGOING);
	if (route_settings) {
		new_settings = psd_settings_merge(settings, route_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge settings with route settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);
		settings = new_settings;
	}

	if (sess->requested_settings) {
		new_settings = psd_settings_merge(settings, route_settings);
		if (!new_settings) {
			psd_err(5, "couldn't merge settings with requested settings");
			goto error_exit_settings;
		}

		psd_settings_free(settings);
		settings = new_settings;
	}

	*ret_settings = settings;

	return 0;

error_exit_settings:
	psd_settings_free(settings);
error_exit:
	return -1;
}
